package com.fanshuai.reactive.reactor.core.jdk;

import lombok.extern.slf4j.Slf4j;
import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;

@Slf4j
public class FlowTest {
    public static class Data {
        public int a;

        public Data(int a) {
            this.a = a;
        }

        @Override
        public String toString() {
            return "Data{" +
                    "a=" + a +
                    '}';
        }
    }

    interface Listener {
        void listen(Data data);
    }

    public static class MySubscriber implements Subscriber<Data> {
        private Listener listener;
        private Subscription subscription;

        public Listener getListener() {
            return listener;
        }

        public void setListener(Listener listener) {
            this.listener = listener;
        }

        @Override
        public void onSubscribe(Subscription subscription) {
            this.subscription = subscription;
            log.info("onSubscribe");
            subscription.request(1);
        }

        @Override
        public void onNext(Data item) {
            listener.listen(item);
            subscription.request(1);
        }

        @Override
        public void onError(Throwable throwable) {
            log.error("error=" + throwable);
            subscription.cancel();
        }

        @Override
        public void onComplete() {
            log.info("onComplete");
        }
    }

    public static class MySubscriber2 implements Subscriber<String> {
        private Subscription subscription;

        @Override
        public void onSubscribe(Subscription subscription) {
            this.subscription = subscription;
            log.info("onSubscribe");
            subscription.request(1);
        }

        @Override
        public void onNext(String item) {
            log.info("MySubscriber2 receive data={}", item);
            subscription.request(1);
        }

        @Override
        public void onError(Throwable throwable) {
            log.error("error=" + throwable);
            subscription.cancel();
        }

        @Override
        public void onComplete() {
            log.info("onComplete");
        }
    }

    public static void main(String[] args) throws Exception {

        MySubscriber subscriber = new MySubscriber();
        subscriber.setListener((Data data) -> {
            try {
                Thread.sleep(1000);
            } catch (Exception e) {

            }
            log.info("receive data={}", data);
        });

        Thread.sleep(3600000);
    }
}
